package broker

import (
	"fmt"
	"github.com/streadway/amqp"
	"go-micro.dev/v4/broker"
	log "manager/collector/logger"
	"manager/utils"
	"sync"
	"time"
)

type publication struct {
	d   amqp.Delivery
	m   *broker.Message
	t   string
	err error
}

func (p *publication) Ack() error {
	return p.d.Ack(false)
}

func (p *publication) Error() error {
	return p.err
}

func (p *publication) Topic() string {
	return p.t
}

func (p *publication) Message() *broker.Message {
	return p.m
}

type AmqpRabbitMq struct {
	//conn           *broker.rabbitMQConn
	addrs       []string
	opts        broker.Options
	queueClient sync.Map
}

func (a *AmqpRabbitMq) Init(opts ...broker.Option) error {
	for _, o := range opts {
		o(&a.opts)
	}
	a.addrs = a.opts.Addrs
	a.queueClient = sync.Map{}
	return nil
}
func (a *AmqpRabbitMq) Options() broker.Options {
	return broker.Options{}
}

func (a *AmqpRabbitMq) Address() string {
	if len(a.addrs) > 0 {
		return a.addrs[0]
	}
	return ""
}
func (a *AmqpRabbitMq) Connect() error {
	return nil
}

func (a *AmqpRabbitMq) Disconnect() error {
	return nil
}
func (a *AmqpRabbitMq) Publish(topic string, m *broker.Message, opts ...broker.PublishOption) error {
	message := m.Body

	// Get the queue client
	var queueClient *Client
	if v, ok := a.queueClient.Load(topic); !ok {
		queueClient = New(topic, a.Address())
		a.queueClient.Store(topic, queueClient)
	} else {
		queueClient = v.(*Client)
	}

	if err := queueClient.Push(message); err != nil {
		log.Errorf("Push failed: %s\n", err)
	}
	return nil
}

func (a *AmqpRabbitMq) Subscribe(topic string, h broker.Handler, opts ...broker.SubscribeOption) (broker.Subscriber, error) {
	var queueClient *Client
	if v, ok := a.queueClient.Load(topic); !ok {
		queueClient = New(topic, a.Address())
		a.queueClient.Store(topic, queueClient)
	} else {
		queueClient = v.(*Client)
	}

	// Give the connection sometime to setup
	<-time.After(time.Second)

	go a.sub(queueClient, topic, h)
	return nil, nil
}

// Subscribe returns a subscriber for the given topic
func (a *AmqpRabbitMq) sub(queueClient *Client, topic string, h broker.Handler) (broker.Subscriber, error) {
	defer utils.ExceptionCatch()

	deliveries, err := queueClient.Consume()
	if err != nil {
		fmt.Printf("Could not start consuming: %s\n", err)
		return nil, fmt.Errorf("could not start consuming: %s", err)
	}

	// This channel will receive a notification when a channel closed event
	// happens. This must be different from Client.notifyChanClose because the
	// library sends only one notification and Client.notifyChanClose already has
	// a receiver in handleReconnect().
	// Recommended to make it buffered to avoid deadlocks
	chClosedCh := make(chan *amqp.Error, 1)
	queueClient.channel.NotifyClose(chClosedCh)

	for {
		select {
		case amqErr := <-chClosedCh:
			// This case handles the event of closed channel e.g. abnormal shutdown
			log.Errorf("AMQP Channel closed due to: %s\n", amqErr)

			deliveries, err = queueClient.Consume()
			if err != nil {
				// If the AMQP channel is not ready, it will continue the loop. Next
				// iteration will enter this case because chClosedCh is closed by the
				// library
				log.Errorf("Error trying to consume, will try again")
				continue
			}

			// Re-set channel to receive notifications
			// The library closes this channel after abnormal shutdown
			chClosedCh = make(chan *amqp.Error, 1)
			queueClient.channel.NotifyClose(chClosedCh)

		case delivery := <-deliveries:
			// This case handles the event of receiving a message
			h(&publication{
				d: delivery,
				m: &broker.Message{Body: delivery.Body},
				t: topic,
			})
			if err := delivery.Ack(false); err != nil {
				log.Errorf("Error acknowledging message: %s\n", err)
			}
		}
	}
}

func (a *AmqpRabbitMq) String() string {
	return ""
}
